#include "common_event.h"
#include "common_event_c.h"

namespace El {
namespace Common {

Event &Event::GetInstance()
{
    static Event instance;
    return instance;
}

Event &Event::GetCustomInstance(std::string instanceName)
{
    static std::mutex g_customInstanceMutex;
    static std::unordered_map<std::string, Event *> instances;

    std::lock_guard<std::mutex> lock(g_customInstanceMutex);
    auto it = instances.find(instanceName);
    if (it == instances.end()) {
        it = instances.emplace(instanceName, new Event()).first;
    }
    return *(it->second);
}

Event::Event()
{
    // 不在构造函数中初始化线程池
    stop_ = true;
}

Event::~Event()
{
    {
        std::unique_lock<std::mutex> lock(queue_mutex_);
        stop_ = true;
    }
    condition_.notify_all();
    for (std::thread &worker : workers_) {
        worker.join();
    }
}

int32_t Event::Subscribe(const std::string &eventName, const Callback &callback, int32_t priority)
{
    auto &manager = managers_[eventName];
    return manager.Register(callback, priority);
}

void Event::Unsubscribe(const std::string &eventName, int32_t handle)
{
    auto it = managers_.find(eventName);
    if (it != managers_.end()) {
        it->second.Unregister(handle);
        if (it->second.IsEmpty()) {
            managers_.erase(it);
        }
    }
}

void Event::Notify(const std::string &eventName, const nlohmann::json &data)
{
    auto it = managers_.find(eventName);
    if (it != managers_.end()) {
        it->second.InvokeAll(eventName, data);
    }
}

std::future<void> Event::NotifyAsync(const std::string &eventName, const nlohmann::json &data)
{
    // 检查并初始化线程池
    bool needInit = false;
    {
        std::unique_lock<std::mutex> lock(queue_mutex_);
        needInit = workers_.empty();
    }

    if (needInit) {
        initializeThreadPool(1); // 默认使用1个工作线程
    }

    auto task = std::make_shared<std::packaged_task<void()>>([this, eventName, data]() { Notify(eventName, data); });
    std::future<void> res = task->get_future();
    {
        std::unique_lock<std::mutex> lock(queue_mutex_);
        if (!stop_) { // 确保在添加任务时线程池还在运行
            tasks_.emplace([task]() { (*task)(); });
            condition_.notify_one();
        }
    }
    return res;
}

bool Event::IsEmpty(const std::string &eventName) const
{
    auto it = managers_.find(eventName);
    return it == managers_.end() || it->second.IsEmpty();
}

void Event::SetThreadPoolSize(size_t size)
{
    initializeThreadPool(size);
}

void Event::initializeThreadPool(size_t size)
{
    // 先停止现有的线程池
    {
        std::unique_lock<std::mutex> lock(queue_mutex_);
        if (!workers_.empty()) {
            stop_ = true;
            condition_.notify_all();
        }
    }

    // 等待所有工作线程结束
    for (std::thread &worker : workers_) {
        if (worker.joinable()) {
            worker.join();
        }
    }

    // 重新初始化线程池
    {
        std::unique_lock<std::mutex> lock(queue_mutex_);
        workers_.clear();
        while (!tasks_.empty()) {
            tasks_.pop();
        }
        stop_ = false;

        for (size_t i = 0; i < size; ++i) {
            workers_.emplace_back(&Event::workerThread, this);
        }
    }
}

void Event::workerThread()
{
    while (true) {
        std::function<void()> task;
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
            condition_.wait(lock, [this] { return stop_ || !tasks_.empty(); });
            if (stop_ && tasks_.empty()) {
                return;
            }
            task = std::move(tasks_.front());
            tasks_.pop();
        }
        task();
    }
}

} // namespace Common
} // namespace El

extern "C" {

int32_t Event_Subscribe(const char *eventName, C_Callback callback, int32_t priority)
{
    auto &event = El::Common::Event::GetInstance();
    std::string eventNameStr = eventName;
    auto cpp_callback = [callback](const std::string &eventName2, const nlohmann::json &data) {
        std::string data_str = data.dump();
        callback(eventName2.c_str(), data_str.c_str());
    };
    return event.Subscribe(eventNameStr, cpp_callback, priority);
}

void Event_Unsubscribe(const char *eventName, int32_t handle)
{
    auto &event = El::Common::Event::GetInstance();
    event.Unsubscribe(eventName, handle);
}

void Event_Notify(const char *eventName, const char *data)
{
    auto &event = El::Common::Event::GetInstance();
    nlohmann::json json_data;
    if (data != nullptr) {
        json_data = nlohmann::json::parse(data, nullptr, false);
    } else {
        json_data = nlohmann::json::object();
    }
    event.Notify(eventName, json_data);
}

void Event_NotifyAsync(const char *eventName, const char *data)
{
    auto &event = El::Common::Event::GetInstance();
    nlohmann::json json_data;
    if (data != nullptr) {
        json_data = nlohmann::json::parse(data, nullptr, false);
        if (json_data.is_discarded()) {
            json_data = nlohmann::json::object();
        }
    } else {
        json_data = nlohmann::json::object();
    }
    // 直接调用NotifyAsync但不等待future的结果
    event.NotifyAsync(eventName, json_data);
}

void Event_SetThreadPoolSize(size_t size)
{
    auto &event = El::Common::Event::GetInstance();
    event.SetThreadPoolSize(size);
}
}